package com.cxf.cxfcommons.zookeeper;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.BackgroundCallback;
import org.apache.curator.framework.api.CuratorEvent;
import org.apache.curator.framework.recipes.locks.InterProcessLock;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.utils.CloseableUtils;
import org.apache.zookeeper.AsyncCallback;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.ACL;
import org.apache.zookeeper.data.Id;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.UnsupportedEncodingException;
import java.util.ArrayList;
import java.util.List;

/**
 * description: ZkUtils <br>
 * date: 2021/11/30 18:43 <br>
 * author: cxf <br>
 * version: 1.0 <br>
 */
public class ZkUtils {

    private static final Logger logger = LoggerFactory.getLogger(ZkUtils.class);

    public static void main(String[] args){
        ZkUtils zkUtils = new ZkUtils();
        // zkUtils.createNode();
       // zkUtils.readNode();
        zkUtils.updateNode();
         // zkUtils.updateNodeAsync();
        //zkUtils.deleteNode();

    }


    public void  createNode(){

        CuratorFramework client = ZkUtils.createSimple("106.13.224.166:2190");
        try {
            client.start();
            String data = "hello node_11 ,i am cxf";
            byte[] payload = data.getBytes("UTF-8");
            String zkoath1 = "/cxf/test/CRUD";
            String zkoath2 = "/cxf/test/CRUD/node-2";
            String zkoath3 = "/cxf/test/CRUD/node-3";
            String zkoath4 = "/cxf/test/CRUD/node-4";

            List<ACL> aclList = new ArrayList<>();
            ACL acl = new ACL();
            acl.setId(new Id("",""));
            aclList.add(acl);

            client.create()
                    .creatingParentsIfNeeded()
                    .withMode(CreateMode.PERSISTENT)
                    .withACL(aclList)
                    .forPath(zkoath1,payload);
/*            client.create()
                    .creatingParentsIfNeeded()
                    .withMode(CreateMode.PERSISTENT_SEQUENTIAL)
                    .forPath(zkoath2,payload);
            client.create()
                    .creatingParentsIfNeeded()
                    .withMode(CreateMode.EPHEMERAL)
                    .forPath(zkoath3,payload);
            client.create()
                    .creatingParentsIfNeeded()
                    .withMode(CreateMode.EPHEMERAL_SEQUENTIAL)
                    .forPath(zkoath4,payload);*/

            Thread.sleep(100000);

        } catch (Exception e) {
            e.printStackTrace();
        }finally {
            CloseableUtils.closeQuietly(client);
        }
    }


    public void  readNode(){

        CuratorFramework client = ZkUtils.createSimple("106.13.224.166:2190");
        try {
            client.start();
            String zkoath1 = "/cxf/test/CRUD/node-1";
            Stat stat = client.checkExists().forPath(zkoath1);
            if(null != stat){
                byte[] bytes = client.getData().forPath(zkoath1);
                String str = new String(bytes, "UTF-8");
                logger.info(">>>>>>>>>>>>>>>>> i get message :"+str);
                String parent = "/cxf";

                List<String> list = client.getChildren().forPath(parent);
                for (String child : list){
                    logger.info(">>>>>>>>>>>>>>>>> i am child : "+child);
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }finally {
            CloseableUtils.closeQuietly(client);
        }
    }


    public void  updateNode(){

        CuratorFramework client = ZkUtils.createSimple("106.13.224.166:2190");
        try {
            client.start();
            String zkoath1 = "/cxf/test/CRUD";
            byte[] payload = "hello cxf CRUD ".getBytes("UTF-8");
            client.setData().forPath(zkoath1,payload);
        } catch (Exception e) {
            e.printStackTrace();
        }finally {
            CloseableUtils.closeQuietly(client);
        }
    }

    public void  updateNodeAsync(){

        CuratorFramework client = ZkUtils.createSimple("106.13.224.166:2190");
        try {
            client.start();

            BackgroundCallback backgroundCallback = new BackgroundCallback() {
                @Override
                public void processResult(CuratorFramework curatorFramework, CuratorEvent curatorEvent) throws Exception {
                    logger.info(">>>>>>>>>>>>>>>"+String.valueOf(curatorFramework.getData()));
                    logger.info(">>>>>>>>>>>>>>>"+String.valueOf(curatorEvent.getData()));
                }
            };

            AsyncCallback.StringCallback asyncCallback = new AsyncCallback.StringCallback() {
                @Override
                public void processResult(int i, String s, Object o, String s1) {
                    logger.info(" >>>>>>>>>i:"+i+"|||||  s:"+s+"||||| o:"+o+"||||| s1:"+s1);
                }
            };

            String zkoath1 = "/cxf/test/CRUD/node-1";
            byte[] payload = "hello cxf async ".getBytes("UTF-8");
            client.setData()
                    .inBackground(backgroundCallback)
                    .forPath(zkoath1,payload);

            Thread.sleep(100000);
        } catch (Exception e) {
            e.printStackTrace();
        }finally {
            CloseableUtils.closeQuietly(client);
        }
    }

    public void  deleteNode(){

        CuratorFramework client = ZkUtils.createSimple("106.13.224.166:2190");
        try {
            client.start();
            String zkoath1 = "/cxf/test/CRUD/node-1";
            byte[] payload = "hello cxf ".getBytes("UTF-8");
            client.delete().forPath(zkoath1);



        } catch (Exception e) {
            e.printStackTrace();
        }finally {
            CloseableUtils.closeQuietly(client);
        }
    }


    public static CuratorFramework createSimple(String connectString){
        // 重试策略 第一次重试等待1s 第二次重试等待2s 第三次重试等待3s
        // baseSleepTimeMs 重试单位 ms
        // maxRetries 重试次数
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        // connectString zk连接地址
        // retryPolicy 重试策略
        return CuratorFrameworkFactory.newClient(connectString, retryPolicy);
    }

    /*
     * @Title: createWithOptions
     * @Description: TODO
     * @Author: cxf
     * @DateTime: 2021/11/30 19:17
     * @param connectString zk连接地址
     * @param retryPolicy 重试策略
     * @param connectionTimeoutMs 连接超时时间
     * @param sessionTimeoutMs 会话超时时间
     * @return org.apache.curator.framework.CuratorFramework 实例
     * @throws
     */
    public static  CuratorFramework createWithOptions(String connectString,RetryPolicy retryPolicy,
                                             int connectionTimeoutMs, int  sessionTimeoutMs){

        return CuratorFrameworkFactory.builder()
                .connectString(connectString)
                .retryPolicy(retryPolicy)
                .connectionTimeoutMs(connectionTimeoutMs)
                .sessionTimeoutMs(sessionTimeoutMs)
                .build();
    }



}
